package consys.social.core.dao;

import org.apache.hadoop.hbase.util.Bytes;
import consys.social.core.bo.Entity;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Delete;
import java.util.Date;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import java.util.ArrayList;
import org.apache.hadoop.hbase.client.Result;
import consys.social.core.bo.DiscussionPost;
import consys.social.core.bo.WallPost;
import consys.social.core.service.SocialUtils;
import java.io.IOException;
import java.util.List;
import org.apache.log4j.Logger;
import consys.social.memcached.dao.spymemcached.Memcached;
import consys.social.memcached.service.spymemcached.WallPostTranscoder;
import consys.social.nosql.dao.hbase.HBase;
import org.apache.hadoop.hbase.client.Put;
import static consys.social.nosql.service.hbase.HBaseHelpers.*;

/**
 *
 * @author Petr Kunc
 */
public class DiscussionDaoImpl implements DiscussionDao {

    private Memcached memcached;
    private ThreadLocal<HBase> walls;
    private ThreadLocal<HBase> discussions;
    private Logger logger = Logger.getLogger(DiscussionDaoImpl.class);
    private int storeSeconds;

    public void setMemcached(Memcached memcached) {
        this.memcached = memcached;
    }

    public void setStoreSeconds(int storeSeconds) {
        this.storeSeconds = storeSeconds;
    }

    public void setDiscussions(ThreadLocal<HBase> walls) {
        this.discussions = walls;
    }

    public void setWalls(ThreadLocal<HBase> walls) {
        this.walls = walls;
    }

    @Override
    public boolean isAuthor(Entity author, WallPost post, DiscussionPost discussion) throws IOException {
        byte[] key = SocialUtils.generateReverseKey(post.getAddedBy().getUid(), post.getDate());
        String stringKey = SocialUtils.generateKey(post.getAddedBy().getUid(), post.getDate());
        Get get = new Get(SocialUtils.generateReverseKey(stringKey, discussion.getDate()));
        addToGet(get, "info", "owner");
        Result res = null;
        try {
            res = discussions.get().get(get);
        } catch (IOException ex) {
            logger.error("Error loading author of discussion from HBase");
            throw ex;
        }
        String authorUid = Bytes.toString(res.value());
        return author.getUid().equals(authorUid);
    }

    @Override
    public void postDiscussion(WallPost post, DiscussionPost discussion) throws IOException {
        byte[] key = SocialUtils.generateReverseKey(post.getAddedBy().getUid(), post.getDate());
        String stringKey = SocialUtils.generateKey(post.getAddedBy().getUid(), post.getDate());
        Put put = new Put(SocialUtils.generateReverseKey(stringKey, discussion.getDate()));
        addToPut(put, "text", "", discussion.getText());
        addToPut(put, "info", "owner", discussion.getAddedBy().getUid());
        try {
            if (discussions.get().add(put)) {
                logger.info("Adding discussion post");
                long value = walls.get().increment(key, "info", "discussionCount", 1);

                WallPost wp = (WallPost) memcached.get(stringKey, new WallPostTranscoder());
                //if wp is null it means it is not on wall anymore...but just for sure
                //we could look at HBase wall backup
                if (wp != null) {
                    wp.setDiscussionCount(value);
                    wp.addDiscussionPost(discussion);
                    memcached.replace(stringKey, wp, storeSeconds, new WallPostTranscoder());
                }
            }
        } catch (IOException ex) {
            logger.error("Unable to perform add on HBase", ex);
            throw ex;
        }
    }

    @Override
    public void deleteDiscussionPost(WallPost post, DiscussionPost discussion) throws IOException {
        byte[] key = SocialUtils.generateReverseKey(post.getAddedBy().getUid(), post.getDate());
        String stringKey = SocialUtils.generateKey(post.getAddedBy().getUid(), post.getDate());
        Delete delete = new Delete(SocialUtils.generateReverseKey(stringKey, discussion.getDate()));
        Get get = new Get(SocialUtils.generateReverseKey(stringKey, discussion.getDate()));
        addToGet(get, "text", "");
        try {
            if (discussions.get().deleteIfGet(get, delete)) {
                logger.info("Deleting discussion post");
                long value = walls.get().increment(key, "info", "discussionCount", -1);
                WallPost wp = (WallPost) memcached.get(stringKey, new WallPostTranscoder());
                //at postPost is similar "problem"
                if (wp != null) {
                    wp.setDiscussionCount(value);
                    wp.deleteDiscussionPost(discussion);
                    memcached.replace(stringKey, wp, storeSeconds, new WallPostTranscoder());
                }
            }
        } catch (IOException ex) {
            logger.error("Unable to perform delete on HBase", ex);
            throw ex;
        }
    }

    @Override
    public List<DiscussionPost> getDiscussion(WallPost post, int count, Date from, boolean newest) throws IOException {
        String key = SocialUtils.generateKey(post.getAddedBy().getUid(), post.getDate());
        List<DiscussionPost> disc = new ArrayList<DiscussionPost>();
        Scan s = new Scan(SocialUtils.generateReverseKey(key, from), SocialUtils.generateReverseKeyMax(key));
        addToScan(s, "text", "");
        addToScan(s, "info", "owner");
        int loaded = 0;
        try {
            ResultScanner rs = discussions.get().getScanner(s);
            if (!newest) {
                rs.next();
            }
            logger.info("Getting discussions");
            for (Result r : rs) {
                loaded++;
                disc.add(new DiscussionLoader().load(r));
                if (loaded >= count) {
                    return disc;
                }
            }
        } catch (IOException ex) {
            logger.error("Getting discussion failed", ex);
            throw ex;
        }
        return disc;
    }
}